package com.xinqi.common.rocketmq.event.consumer;

import com.xinqi.common.core.enums.RocketMqEnum;
import com.xinqi.common.core.utils.SpringUtils;
import com.xinqi.common.rocketmq.dto.RocketMqBean;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientConfigurationBuilder;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.StaticSessionCredentialsProvider;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.PushConsumer;

import java.util.Collections;

/**
 * 切片消费消息
 *
 * @author dzy
 */
@Slf4j
public class DemoConsumerListener {

    private static final RocketMqBean rocketMqBean = SpringUtils.getBean(RocketMqBean.class);

    public static void pushConsumerMsg() throws ClientException {

        final ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(rocketMqBean.getEndpoints());
        /**
         * 如果是使用公网接入点访问，configuration还需要设置实例的用户名和密码。用户名和密码在控制台实例详情页获取。
         * 如果是在阿里云ECS内网访问，无需填写该配置，服务端会根据内网VPC信息智能获取。
         */
        //builder.setCredentialProvider(new StaticSessionCredentialsProvider("Instance UserName", "Instance Password"));
        ClientConfiguration clientConfiguration =
            builder.setCredentialProvider(new StaticSessionCredentialsProvider(rocketMqBean.getUsername(), rocketMqBean.getPassword()))
                .build();
        //订阅消息的过滤规则，表示订阅所有Tag的消息。
        String tag = RocketMqEnum.SLICE.getCode();
        FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
        //初始化PushConsumer，需要绑定消费者分组ConsumerGroup、通信参数以及订阅关系。
        PushConsumer pushConsumer = provider.newPushConsumerBuilder()
            .setClientConfiguration(clientConfiguration)
            //设置消费者分组。
            .setConsumerGroup(RocketMqEnum.SLICE.getCode())
            //设置预绑定的订阅关系。
            .setSubscriptionExpressions(Collections.singletonMap(rocketMqBean.getTopic(), filterExpression))
            //设置消费监听器。
            .setMessageListener(messageView -> {
                log.info("Consume Message: " + messageView);
                return ConsumeResult.SUCCESS;
            }).build();
    }

}
